ref(consumer): remove the BLQ (backlog-queue) mechanism - #8166

> **Stacked on #8166** (BLQ removal). Base is `ref/remove-blq`; review
that first. Retarget to `master` once #8166 merges.

## Summary

Adds `DlqByAge`, a stateless arroyo strategy for the Rust consumer that
sheds load during a backlog. When enabled, it routes a **configurable
proportion** of messages older than a threshold (by **Kafka broker
timestamp**) to the DLQ via the standard `InvalidMessage` mechanism. The
rest are forwarded untouched — so a backlogged consumer catches up to
fresh data while the parked backlog is backfilled from the DLQ later.

This replaces the removed BLQ mechanism with something far simpler: **no
state machine, no consumer panic, no separate producer.** It leans
entirely on arroyo's existing DLQ buffer + policy.

## How it works

- Sits at the front of the pipeline (where `BLQRouter` was), *before*
decode/process, so dropped messages cost almost nothing.
- For each message: if the per-storage sample rate is 0 → forward (fast
path). Else if `now - broker_ts <= threshold` → forward. Else roll `rand
< rate`: hit → `Err(SubmitError::InvalidMessage)` (→ DLQ); miss →
forward.
- **Only inserted when a DLQ topic is configured** (`dlq_configured`
flag from `consumer.rs`). Without a DLQ policy, arroyo logs an
`InvalidMessage` and silently drops it — which would lose data — so we
never wire the strategy in that case.

## Kept out of Sentry

Age-based routing is an expected load-shedding outcome, not an error.
Dead-lettered messages use `InvalidMessageReason::Ignored`, so arroyo
logs them at DEBUG (not ERROR) and the tracing→Sentry layer
(`logging.rs`) does **not** turn them into Sentry issues. Both reasons
produce to the DLQ identically — only the log level differs. This is the
same intent as the `SilencedDLQMessage` path from #8012.

## Configuration (sentry-options, read per-message, runtime-tunable)

| Option | Type | Default | Meaning |
|---|---|---|---|
| `consumer.dlq_by_age_sample_rate_by_storage` | dict `{storage:
number}` | `{}` | fraction (0–1) of too-old messages to dead-letter; no
entry ⇒ 0 (disabled) |
| `consumer.dlq_by_age_threshold_seconds` | int | `3600` | age cutoff by
broker timestamp |

**Ships disabled** — no storage is enabled and the rate defaults to 0,
so this is an inert passthrough until an operator opts a storage in.

Metrics: `dlq_by_age.stale_seen` and `dlq_by_age.routed_to_dlq` (storage
is already a global metric tag).

## Test plan
- `cargo test --lib dlq_by_age` — 6 unit tests pass:
disabled-by-default, explicit rate 0, fresh-forwarded, rate 1 routes all
stale, per-storage scoping, and a partial-rate split. Each DLQ'd message
is asserted to carry `Ignored`.
- `cargo clippy --all-targets` clean; `cargo fmt --check` clean.

## Follow-up (PR 3, ops)
Enable per storage via sentry-options and watch
`dlq_by_age.routed_to_dlq` + DLQ topic lag.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>